package com.zf.canal.demo;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.zf.canal.config.CanalConnectProperties;
import com.zf.canal.handler.RowDataHandler;
import com.zf.canal.utils.TableHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 基于 spring 实现 canal 定时同步数据
 *
 * @author Zhou Feng
 */
@Component
public class SpringCanalSyncClientDemo {
    private static final Logger logger = LoggerFactory.getLogger(SpringCanalSyncClientDemo.class);
    @Autowired
    private CanalConnector connector;
    @Autowired
    private CanalConnectProperties canalConnectProperties;


    /**
     * 定时同步数据，fixedDelay 表示在 上一次任务执行结束 到 下一次任务开始 之间 间隔 2s，
     * 注意是在上一个任务执行结束后才会开始等待执行下一次的任务执行，如果上一次任务没有结束，则不会开始下一次的等待
     */
    @Scheduled(fixedDelay = 2000)
    public void sync() {
        // 打开连接
        CanalConnector connector = openConnector();
        try {
            // 获取指定条数的日志数据，具体数量以实际为准，小于等于 BatchFetchSize
            Message message = connector.getWithoutAck(canalConnectProperties.getBatchFetchSize());
            //获取批次ID
            long batchId = message.getId();
            //这个批次取到了多少数据
            int size = message.getEntries().size();
            try {
                if (batchId != -1 && size > 0) {
                    //如果有数据, 就处理数据
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        // 只处理 ROWDATA 类型的 binlog
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            handlerEntry(entry);
                        }
                    }
                }
                //处理完成后，进行 batchId 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                connector.ack(batchId);
            } catch (Exception e) {
                logger.error("Canal异常, Message: {}; \n异常原因: {}", message, e.toString());
                // 直接ack, 防止一直出错造成死循环，这些操作产生的数据就会丢失，只能通知到人，手动处理
                connector.ack(batchId);
            }
        } catch (Exception e) {
            logger.error("Canal定时同步任务异常: {}", e.toString());
        }
    }


    /**
     * 创建一个访问canal的连接
     */
    public CanalConnector openConnector() {
        //打开连接
        connector.connect();
        //订阅哪些数据库的哪些表的变化，正则表达式，多个表达式用 "," 分隔
        connector.subscribe(canalConnectProperties.getFilterRegex());
        //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
        connector.rollback();
        return connector;
    }


    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    private void handlerEntry(CanalEntry.Entry entry) {
        //RowChange对象，包含了sql影响到的行数据，以及sql类型等相关信息
        CanalEntry.RowChange rowChange;
        try {
            rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("CanalEntry 转换为 RowChange 失败: " + entry.toString(), e);
        }
        // 表名
        String tableName = entry.getHeader().getTableName();
        // 表明对应的数据处理器
        RowDataHandler rowDataHandler = TableHelper.getTableHandler(tableName);
        // 表名对应的实体类型
        Class<?> entityClass = TableHelper.getTableEntityClass(tableName);
        // 操纵类型
        CanalEntry.EventType eventType = rowChange.getEventType();
        //获取RowChange对象里的每一行数据，只有增删改操作才会有 RowData 行数据，其他操作的getRowDatasList是空list
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            // 只要有rowData 数据，那么 getBeforeColumnsList 和 getAfterColumnsList 一定是 >= 0 的list，不会为null
            Object before = convertRowDataToEntity(rowData.getBeforeColumnsList(), entityClass);
            Object after = convertRowDataToEntity(rowData.getAfterColumnsList(), entityClass);
            switch (eventType) {
                case DELETE:
                    rowDataHandler.delete(before);
                    break;
                case INSERT:
                    rowDataHandler.insert(after);
                    break;
                case UPDATE:
                    rowDataHandler.update(before, after);
                    break;
            }
        }
    }

    /**
     * 将 rowData数据转换成 对应的实体类
     */
    private Object convertRowDataToEntity(List<CanalEntry.Column> columns, Class<?> entityClass) {
        JSONObject jsonObject = new JSONObject();
        for (CanalEntry.Column column : columns) {
            // 下划线转驼峰
            String name = Arrays.stream(StringUtils.split(column.getName().toLowerCase(), '_'))
                    .map(StringUtils::capitalize)
                    .collect(Collectors.joining(""));
            name = Character.toLowerCase(name.charAt(0)) + name.substring(1);
            jsonObject.put(name, column.getValue());
        }
        return jsonObject.to(entityClass);
    }

}
